package com.atguigu.bigdata.gmall.realtime.app.dwd;/**
 * @Classname Dwd_03_Dwd_VideoPlay
 * @Description TODO
 * @Date 2022/10/25 18:27
 * @Created by 86152
 */

import com.atguigu.bigdata.gmall.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.gmall.realtime.common.Constant;
import com.atguigu.bigdata.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author: Lime
 * @description 视频播放
 * @date: 2022/10/25 18:27
 */
public class Dwd_03_Dwd_VideoPlay extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_03_Dwd_VideoPlay().init(
                3004,
                2,
                "Dwd_03_Dwd_VideoPlay"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        //1.获取ods_db数据
        readOdsDb(tEnv, "Dwd_03_Dwd_VideoPlay");

        //2.过滤出章节进度表数据11
        Table videoPlay = tEnv.sqlQuery("select\n" +
                " data['id'] id,\n" +
                " data['course_id'] course_id,\n" +
                " data['chapter_id'] chapter_id,\n" +
                " data['user_id'] user_id,\n" +
                " data['position_sec'] position_sec,\n" +
                " data['create_time'] create_time,\n" +
                " pt,\n" +
                " ts\n" +
                "from ods_db\n" +
                "where `database`='gmall2022' and `table`='user_chapter_process' and `type`='insert'");
        tEnv.createTemporaryView("videoPlay",videoPlay);
//        tEnv.executeSql("select * from videoPlay").print();

        //5.写出到kafka
        tEnv.executeSql("create table dwd_video_play(\n" +
                " id string,\n" +
                " course_id string,\n" +
                " chapter_id string,\n" +
                " user_id string,\n" +
                " position_sec string,\n" +
                " create_time string,\n" +
                " pt timestamp_ltz(3),\n" +
                " ts bigint,\n" +
                " primary key(id) not enforced\n" +
                ")" + SQLUtil.getUpsertKafkaSink(Constant.TOPIC_DWD_VIDEO_PLAY));

        tEnv.executeSql("insert into dwd_video_play select * from videoPlay");


    }
}
